构造方法以及参数解析
先来看一下 ThreadPoolExecutor
的几个构造方法:
- ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue
workQueue) - ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue
workQueue,ThreadFactory threadFactory) - ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue
workQueue,RejectedExecutionHandler handler) - ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue
workQueue,ThreadFactory threadFactory, RejectedExecutionHandler handler)
下面来解释一下这结果参数的意义:
- corePoolSize:核心线程的数量。
- 核心线程默认是没有超时的,也就是说就算线程闲置,也不会被处理。但是如果设置了
allowCoreTimeOut
为true,那么当核心线程闲置时,会被回收。 - 当提交一个任务到线程池时,线程池会创建一个线程来执行任务,即使其他空闲的核心线程能够执行新任务也会创建线程,等到需要执行的任务数大于线程池核心线程的数量时就不再创建。如果调用了线程池的
prestartAllCoreThreads
方法,线程池会提前创建并启动所有核心线程。 - 核心线程在创建 Worker 的时候会赋值 firstTask。
- 核心线程默认是没有超时的,也就是说就算线程闲置,也不会被处理。但是如果设置了
- maximumPoolSize:线程池的允许创建的最大线程数量,被
CAPACITY
限制,最大职能是 2^29-1。- 当 corePoolSize =< 线程数 < maximumPoolSize,且任务队列已满时。线程池会创建新线程来处理任务。
- 当 maximumPoolSize < 线程数时,且任务队列已满时,线程池会拒绝处理任务而抛出异常。
- keepAliveTime:线程活动保持时间。线程池的工作线程空闲后,保持存活的时间。
- unit:线程活动保持时间的单位。
- workQueue:用于保存等待执行的任务的阻塞队列。
- threadFactory:设置创建线程的工厂。可用于统一设置线程的一些属性。
- handler:线程池的拒绝策略。当队列和线程池都满了,说明线程池处于饱和状态,那么必须采取一种策略处理提交的新任务。下面介绍源码中给出的四种实现。
- AbortPolicy:直接抛出
RejectedExecutionException
异常。这是ThreadPoolExecutor
的默认策略。 - CallerRunsPolicy:直接在调用者所处的线程来执行任务。
- DiscardOldestPolicy:丢弃队列中最早创建的任务,并尝试重新触发执行一次线程池任务。
- DiscardPolicy:直接丢弃该任务,不做任何处理。
- 也可以根据应用场景需要来实现
RejectedExecutionHandler
接口自定义策略。
- AbortPolicy:直接抛出
概念解释
任务队列
任务队列就是用于保存等待执行的任务队列。
通过前面的介绍我们可以了解到线程池创建线程的流程:
- 当线程数 < corePoolSize 时,线程池会创建一个线程来执行任务,即使其他空闲的核心线程能够执行新任务也会创建线程。
- 当 corePoolSize =< 线程数 < maximumPoolSize,且任务队列已满时。线程池会创建新线程来处理任务。
- 当 maximumPoolSize < 线程数时,且任务队列已满时,线程池会拒绝处理任务而抛出异常。
因此,任务队列的类型也会影响到线程创建逻辑。
在源码中任务队列的类型是:
1 | private final BlockingQueue<Runnable> workQueue; |
通过实现 BlockingQueue
接口,我们可以自定义任务队列。
- SynchronousQueue:
Executors.newCachedThreadPool()
创建的是这种类型的队列。它也是无界队列。它的特点是在某次添加任务后必须等待其他线程取走后才能继续添加。 - LinkedBlockingQueue:
Executors.newSingleThreadExecutor
和Executors.newFixedThreadPool
创建的是这种类型的队列。它也是无界队列。使用这种队列有可能造成会堆积大量的请求,从而导致 OOM。 - DelayedWorkQueue:
Executors.newScheduledThreadPool
创建的是这种类型的队列。 - ArrayBlockingQueue:它是有界队列,队列的最大值可以指定,有助于防止资源耗尽。
Worker
Worker 就是线程池中任务的执行者,它的数量为 corePoolSize 的大小。
Worker 相当于对任务和线程的一层包装,它可以控制一些在线程执行过程中的中断操作。
Worker 继承自 AbstractQueuedSynchronizer
并且实现了 Runnable
。
1 | private final class Worker |
其实核心方法就是 runWorker(Worker w)
方法,这个方法会在后面的源码分析中介绍。
源码解析
当 ThreadPoolExecutor
创建好后,就会调用 execute(Runnable command)
方法来执行线程。
1 | public void execute(Runnable command) { |
下面再来看一下 addWorker
方法,该方法的主要作用就是
创建Worker
1 | private boolean addWorker(Runnable firstTask, boolean core) { |
执行任务
执行任务是通过 Worker 的 run 方法发起的,核心方法就是 runWorker(Worker w)
:
1 | final void runWorker(Worker w) { |
runWorker(Worker w)
方法的任务就是来执行线程池中的任务,对于核心线程来说,当 firstTask 执行完毕后,它还会通过 getTask()
方法获取等待队列中的任务来执行,该方法会根据 allowCoreThreadTimeOut
的配置和当前核心线程的数量来决定是否阻塞,根据 keepAliveTime
的时间来决定阻塞多久。
下面就来看一下 getTask()
方法。
1 | private Runnable getTask() { |